#import absl
import logging
logging.basicConfig(level=logging.DEBUG)
import ray
import calc_manager
from calc_manager.base.runnable import Runnable, runnable_task
from calc_manager.executor.exec import *
from calc_manager.operator import Operator, OperatorInfo, print_operator_info
from calc_manager.ray.ray_manager import RayManager
from calc_manager.ray.remote_task import RayRemoteRunnableTaskFatory
from calc_manager.task.localtask import DefaultTaskFactory


local_operator_info_args = {
    "operator_name": "test_sample_plugin",
    "operator_path": "operators_sample/sample_plugin/plugin.py",
    "operator_func_entrypoint": "plugin",
    "operator_desc": "This is a test plugin operator",
}
local_operator_info = OperatorInfo(**local_operator_info_args)
remote_local_info_args = {
    "operator_name": "test_sample_plugin",
    "operator_path": "sample_plugin/plugin.py",
    "operator_func_entrypoint": "plugin",
    "operator_desc": "This is a test plugin operator",
}
remote_operator_info = OperatorInfo(**remote_local_info_args)


from calc_manager.base.plugins import PluginInfo
from calc_manager.plugins.plugins_manager import PluginManager
from calc_manager.base.fields import FieldInfo, REQUIRED_KEY

plugin_info = PluginInfo(
    plugin_name="csv_inner",
    plugin_class_name='CsvPlugin',
    plugin_module_name='calc_manager.plugins_inner.csv_inner',
    plugin_class_param_name='csv_plugin',
    plugin_description='csv test',
    plugin_fields={
        "field1": FieldInfo(name="osspath", type="str", optional_field=REQUIRED_KEY, default_val=None, description="osspath"),
    }
)
pm = PluginManager()
pm.register_plugin(plugin_info)
plugins_argv={"csv_inner": {"osspath": "https://oss.de3.cc:62135/public/example.csv"}}

default_exec = DefaultExecutor()
local_task_result = exec_operator_task(
    default_exec,
    local_operator_info,
    DefaultTaskFactory(pm, plugins_argv),
    param={"k": 4},
)

manager = RayManager(
    "ray://10.130.9.3:10001",
    {
        "runtime_env": {
            "py_modules": [calc_manager],
            "working_dir": "./operators_sample",
            # "pip": ["emoji", "numpy", "pynvim"],
        }
    },
)

remote_task_result = exec_operator_task(
    manager,
    remote_operator_info,
    RayRemoteRunnableTaskFatory(pm, plugins_argv),
    param={"k": 4},
)

print(f"local task result {local_task_result}")
print(f"remote task result {remote_task_result}")